package br.unifor.cmi.ha.jgcs;


import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Proxy;
import java.net.SocketAddress;
import java.util.LinkedList;
import java.util.NoSuchElementException;

import net.sf.jgcs.ClosedSessionException;
import net.sf.jgcs.DataSession;
import net.sf.jgcs.GroupConfiguration;
import net.sf.jgcs.JGCSException;
import net.sf.jgcs.NotJoinedException;
import net.sf.jgcs.Protocol;
import net.sf.jgcs.ProtocolFactory;
import net.sf.jgcs.Service;
import net.sf.jgcs.membership.BlockListener;
import net.sf.jgcs.membership.BlockSession;
import net.sf.jgcs.membership.MembershipSession;
import net.sf.jgcs.utils.FactoryUtil;

import org.ow2.cmi.ha.HaMessageData;
import org.ow2.cmi.ha.MessageManager;
import org.ow2.cmi.ha.ReplicationManager;
import org.ow2.util.log.Log;
import org.ow2.util.log.LogFactory;

import br.unifor.g2cl.G2CLMessage;
import br.unifor.g2cl.GroupRequest;
import br.unifor.g2cl.IMarshalDataSession;
import br.unifor.g2cl.MarshalDataSessionFactory;
import br.unifor.g2cl.Util;
import br.unifor.g2cl.hl.IMessageDispatcher;
import br.unifor.g2cl.hl.MessageDispatcherListener;
import br.unifor.g2cl.hl.RspList;
import br.unifor.util.cluster.jgcs.ConnectionManager;
import br.unifor.util.cluster.jgcs.MessageDispatcherWrapper;

/**
 * This class processes the HA messages received through JGCS.
 * @author Leandro Sales - leandro.shp@gmail.com
 */
public class JGCSMessageManager implements MessageDispatcherListener, MessageManager, BlockListener{

    /**
     * Logger.
     */
    private static Log logger = LogFactory.getLog(JGCSMessageManager.class);

    private final ReplicationManager replicationManager;

    /**
     * JGCS related classes.
     */
    private MembershipSession controlSession;
    private IMarshalDataSession dataSession;

    private IMessageDispatcher dispatcher;

    private SocketAddress address;

    private LinkedList<HaMessageData> messages;

    private ProcessThread processThread;

    private long messagesNumber;

    private long messagesSize;

    private final String jgcs_properties;

    private final String jgcsHAGroupName;
    
    /**
     * Constructor.
     */
    public JGCSMessageManager(	final String jgcs_properties,
					            final ReplicationManager replicationManager,
					            final long timeout) {
    	     
        this.replicationManager = replicationManager;
        this.jgcs_properties = jgcs_properties;
    	
        String jgcsHAGroupName = "";
        
        try {
        	
        	FactoryUtil jgcsConf = new FactoryUtil(jgcs_properties);
            
            ProtocolFactory o = (ProtocolFactory) jgcsConf.getInstance("jgcsProtocol");
            Protocol p = o.createProtocol();
            GroupConfiguration group = (GroupConfiguration) jgcsConf.getInstance("jgcsGroup");
            Service service = (Service) jgcsConf.getInstance("jgcsService");
            
            MarshalDataSessionFactory marshalFactory = (MarshalDataSessionFactory) jgcsConf.getInstance("jgcsMarshalDataSession");
        	        	
            jgcsHAGroupName = group.toString();

            DataSession dataSession = p.openDataSession(group);
            controlSession = (MembershipSession)p.openControlSession(group);
            
            if(controlSession instanceof BlockSession){
            	BlockSession blockSession = (BlockSession)controlSession;
            	blockSession.setBlockListener(this);
            }
            
            logger.info("jgcs HA group name is: " + jgcsHAGroupName);

            logger.info("jgcs HA jndi properties is: " + jgcs_properties);
        	
            this.dataSession = marshalFactory.getMarshalDataSession(controlSession, dataSession);
            
            IMessageDispatcher messageDispatcher = new MessageDispatcherWrapper(this.dataSession,controlSession, service);
            messageDispatcher.setMessageDispatcherListener(this);
            messageDispatcher.setLocal(false);
            
            ConnectionManager connectionManager = new ConnectionManager(timeout, messageDispatcher, IMessageDispatcher.class);
            connectionManager.setControlSession(controlSession);
            messageDispatcher.setMembershipListener(connectionManager);
            
            controlSession.join();

            dispatcher = (IMessageDispatcher) Proxy.newProxyInstance(IMessageDispatcher.class.getClassLoader(), new Class[]{IMessageDispatcher.class}, connectionManager);
            
	        messages = new LinkedList<HaMessageData>();

	        processThread = new ProcessThread(messages);
	        processThread.setDaemon(true);
	        processThread.start();
	        
	        address = controlSession.getLocalAddress();
	        logger.debug(address + " joined group " + jgcsHAGroupName);

	        // Initialize statistics
	        messagesNumber = 0;
	        messagesSize = 0;
	        
		} catch (ClosedSessionException e) {
			e.printStackTrace();
		} catch (JGCSException e) {
			e.printStackTrace();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		this.jgcsHAGroupName = jgcsHAGroupName;
        

    }

	/* (non-Javadoc)
     * @see org.ow2.cmi.ha.MessageManager#sendMessage(org.ow2.cmi.ha.HaMessageData)
     */
    public void sendMessage(final HaMessageData messageData) {

    	try {
    		
			G2CLMessage message = (G2CLMessage)dataSession.createMessage();
			byte[] arr = Util.getArrayFromObject(messageData);
			message.setPayload(arr);
			
			SocketAddress localAddress = dispatcher.getControlSession().getLocalAddress();
			
			message.setSenderAddress(localAddress);
			
			// Compute statistics
			messagesNumber++;
			messagesSize += arr.length;

			RspList rspList = dispatcher.castMessage(null, message, GroupRequest.GET_FIRST, 0);
			if (logger.isDebugEnabled()) {
			    
			    logger.debug("Message sended. Length: " + arr.length + " From: " + localAddress);
			    logger.debug("Response: " + rspList.toString());
			}
		} catch(Exception e) {
			logger.error("error sending message", e);
		}
    }
	/*
      Processes the replication messages received. Depending on the message's type
     * a concrete process will be performed
     */
    public Object handle(final G2CLMessage message) {
        if (logger.isDebugEnabled()) {
            logger.debug("-----------------------------------------------");
            logger.debug("Processing message: " + message + " in: " + dispatcher.getControlSession().getLocalAddress());

        }

        
		try {
			Object o = Util.getObjectFromByte(message.getPayload(),0,message.size());
			if (o instanceof HaMessageData) {
			    synchronized (messages) {
			        messages.add((HaMessageData) o);
			        // COMPLETE: Check if this should be > 0
			        if (messages.size() == 1) {
			            processThread.resumeExecution();
			        }
			    }
			} else {
			    logger.debug("\tNo action performed, unknown message format!!!");
			}
			logger.debug("-----------------------------------------------");
		} catch (Exception e) {
			logger.error("error receiving message", e);
		}
        return null;
    }

    /* (non-Javadoc)
     * @see org.ow2.cmi.ha.MessageManager#clear()
     */
    public void clear() {
        dispatcher.stop();
        
        try {
			controlSession.leave();
		} catch (ClosedSessionException e) {
			e.printStackTrace();
		} catch (JGCSException e) {
			e.printStackTrace();
		}
		
        processThread.stopExecution();
    }

    /**
     * Get the number of replicated messages sent.
     * @return the number of replicated messages
     */
    public long getNumberofReplicatedMessages() {
        return messagesNumber;
    }

    /**
     * Get the average size of the replicated messages sent.
     * @return the average size of the replicated messages sent.
     */
    public double getAvgSizeofReplicatedMessages() {
        if (messagesNumber > 0) {
            return messagesSize / messagesNumber;
        } else {
            return 0;
        }
    }

    /**
     * Get the total size of the replicated messages sent.
     * @return the total size of the replicated messages
     */
    public double getTotSizeofReplicatedMessages()  {
        return messagesSize;
    }

    /**
     * This Thread process messages in messages list.
     * @author Francisco Perez-Sorrosal (fpsorrosal@no-spam@fi.upm.es)
     * @author Alberto Paz-Jimenez (apaz@no-spam@fi.upm.es)
     */
    private final class ProcessThread extends Thread {
        private volatile Thread test;
        private final LinkedList<HaMessageData> messages;
        private volatile boolean suspended;

        ProcessThread(final LinkedList<HaMessageData> messages) {
            this.messages = messages;
        }

        @Override
        public void run() {
            logger.debug("ProcessMessage thread started");

            suspended = false;
            test = Thread.currentThread();
            while (test == Thread.currentThread()) {
                try {
                    HaMessageData o = null;
                    synchronized (messages) {
                        if (messages.size() > 0) {
                            o = messages.removeFirst();
                        } else {
                            suspendExecution();
                        }
                    }

                    if (o != null) {
                        replicationManager.processMessage(o);
                    }
                } catch (NoSuchElementException e) {
                    // If there are no messages suspend the thread
                    suspendExecution();
                }

                if (suspended) {
                    synchronized(this) {
                        try {
                            while (suspended && test == Thread.currentThread()) {
                                wait();
                            }
                        } catch (InterruptedException e1) {
                            ;
                        }
                    }
                }
            }
        }

        void stopExecution() {
            logger.debug("ProcessMessage thread stoped");

            synchronized (this) {
                test = null;
                notify();
            }
        }

        void suspendExecution() {
            logger.debug("ProcessMessage thread suspended");

            synchronized (this) {
                suspended = true;
            }
        }

        void resumeExecution() {
            logger.debug("ProcessMessage thread resumed");

            synchronized (this) {
                suspended = false;
                notify();
            }
        }
    }

	public String jgroupsConfFileName() {
		throw new UnsupportedOperationException("YOU'RE USING JGCS!!");
	}

	public void onBlock() {
		
		BlockSession blockSession = (BlockSession)controlSession;
		try {
			blockSession.blockOk();
		} catch (NotJoinedException e) {
			e.printStackTrace();
		} catch (JGCSException e) {
			e.printStackTrace();
		}
		
	}

}

